package com.aceitunaproject.server.test;

import java.io.IOException;
import java.math.BigInteger;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.aceitunaproject.server.ServerConfiguration;
import com.aceitunaproject.server.player.PlayerService;
import com.aceitunaproject.shared.message.Message;
import com.aceitunaproject.shared.message.MessageConstants;
import com.sun.sgs.app.ClientSession;
import com.sun.sgs.auth.Identity;
import com.sun.sgs.impl.io.SocketEndpoint;
import com.sun.sgs.impl.io.TransportType;
import com.sun.sgs.impl.sharedutil.MessageBuffer;
import com.sun.sgs.io.Connection;
import com.sun.sgs.io.ConnectionListener;
import com.sun.sgs.io.Connector;
import com.sun.sgs.kernel.KernelRunnable;
import com.sun.sgs.kernel.TransactionScheduler;
import com.sun.sgs.protocol.simple.SimpleSgsProtocol;
import com.sun.sgs.service.DataService;
import com.sun.sgs.test.util.SgsTestNode;

/**
 * Generic SGS client. This class is a port of the SGS client implementations
 * provided in the service test cases for Darkstar.
 * 
 * @author Maciej Miechowicz
 */
public class SgsTestClient {

	private static final int WAIT_TIME = 3000;
	private static final String LOGIN_FAILED_MESSAGE = "login failed";

	private String name;
	private Connector<SocketAddress> connector;
	private ConnectionListener listener;
	private Connection connection;
	private boolean connected = false;
	private final Object lock = new Object();

	private final Object binarySessionMessageLock = new Object();
	private final Object objectSessionMessageLock = new Object();
	private final Object binaryChannelMessageLock = new Object();
	private final Object objectChannelMessageLock = new Object();

	private boolean loginAck = false;
	private boolean loginSuccess = false;
	private boolean loginRedirect = false;
	private boolean logoutAck = false;
	private boolean awaitGraceful = false;
	private Map<BigInteger, String> channelIdToName =
		new HashMap<BigInteger, String>();
	private Map<String, BigInteger> channelNameToId =
		new HashMap<String, BigInteger>();
	private String redirectHost;
	private int redirectPort;
	private final List<ByteBuffer> binarySessionMessages =
		new ArrayList<ByteBuffer>();
	private final List<Message> objectSessionMessages =
		new ArrayList<Message>();
	private final List<BinaryChannelMessage> binaryChannelMessages =
		new ArrayList<BinaryChannelMessage>();
	private final List<ObjectChannelMessage> objectChannelMessages =
		new ArrayList<ObjectChannelMessage>();

	private DataService dataService;
	private TransactionScheduler trxScheduler;
	private Identity owner;

	/**
	 * Creates the client instance
	 */
	public SgsTestClient(SgsTestNode testNode) {
		dataService = testNode.getDataService();
		trxScheduler = testNode.getSystemRegistry().getComponent(TransactionScheduler.class);
		owner = testNode.getProxy().getCurrentOwner();
	}

	/**
	 * Connects to the server on the given port
	 */
	public boolean connect(String host, int port)
	{
		if (connected)
			throw new RuntimeException("Already connected");

		connected = false;
		listener = new Listener();
		try {
			SocketEndpoint endpoint =
				new SocketEndpoint(
						new InetSocketAddress(host, port),
						TransportType.RELIABLE);
			connector = endpoint.createConnector();
			connector.connect(listener);
		} catch (Exception ex) {
			throw new RuntimeException("Error connecting to server", ex);
		}

		synchronized (lock) {
			try {
				if (connected == false)
					lock.wait(WAIT_TIME * 2);

				if (connected != true)
					throw new RuntimeException(
					"Server connection timeout");

			} catch (InterruptedException e) {
				throw new RuntimeException(
						"Server connection timeout", e);
			}
		}

		return true;
	}

	/**
	 * Indicates if the client is connected
	 */
	public boolean isConnected() {
		synchronized (lock) {
			return connected;
		}
	}

	/**
	 * Disconnects the client
	 */
	public void disconnect() {
		synchronized (lock) {
			if (! connected)
				return;

			try {
				connection.close();
				lock.wait(WAIT_TIME);
			} catch (Exception e) {
				lock.notifyAll();
				return;
			} finally {
				if (connected)
					reset();
			}
		}
	}

	/**
	 * Resets the client state
	 */
	private void reset() {
		assert Thread.holdsLock(lock);
		connected = false;
		connection = null;
		loginAck = false;
		loginSuccess = false;
		loginRedirect = false;
		resetMessageQueues();
	}

	/**
	 * Logs a user in
	 */
	public void login(String user, String pass) {
		synchronized (lock) {
			if (connected == false)
				throw new RuntimeException("Not connected");
		}

		this.name = user;

		MessageBuffer buf =
			new MessageBuffer(2 + MessageBuffer.getSize(user) +
					MessageBuffer.getSize(pass));
		buf.putByte(SimpleSgsProtocol.LOGIN_REQUEST).
		putByte(SimpleSgsProtocol.VERSION).
		putString(user).
		putString(pass);
		loginAck = false;

		try {
			connection.sendBytes(buf.getBuffer());
		} catch (IOException e) {
			throw new RuntimeException(e);
		}

		String host = null;
		synchronized (lock) {
			try {
				if (loginAck == false) {
					lock.wait(WAIT_TIME * 2);
				}
				if (loginAck != true) {
					throw new RuntimeException(
							"DummyClient.login[" + name + "] timed out");
				}
				if (loginSuccess) {
					return;
				} else if (loginRedirect) {
					host = redirectHost;
				} else {
					throw new RuntimeException(LOGIN_FAILED_MESSAGE);
				}
			} catch (InterruptedException e) {
				throw new RuntimeException(
						"DummyClient.login[" + name + "] timed out", e);
			}
		}

		// cache a local copy of redirect port, in case it's ever
		// cleared by disconnect
		int port = redirectPort;
		disconnect();
		connect(host, port);
		login(user, pass);
	}

	/**
	 * Logs a user out
	 */
	public void logout() {
		synchronized (lock) {
			if (connected == false)
				return;

			logoutAck = false;
			awaitGraceful = true;
		}

		MessageBuffer buf = new MessageBuffer(1);
		buf.putByte(SimpleSgsProtocol.LOGOUT_REQUEST);
		try {
			connection.sendBytes(buf.getBuffer());
		} catch (IOException e) {
			throw new RuntimeException(e);
		}

		synchronized (lock) {
			try {
				if (logoutAck == false)
					lock.wait(WAIT_TIME);

				if (logoutAck != true)
					throw new RuntimeException("Disconnect timed out");
			} catch (InterruptedException e) {
				throw new RuntimeException("Disconnect timed out", e);
			} finally {
				if (! logoutAck)
					disconnect();
			}
		}
	}

	/**
	 * Throws a {@code RuntimeException} if this session is not
	 * logged in.
	 */
	private void checkLoggedIn() {
		synchronized (lock) {
			if (!connected || !loginSuccess) {
				throw new RuntimeException("Not connected or loggedIn");
			}
		}
	}

	/**
	 * Sends the given binary message to the server
	 * 
	 * @param message message to send
	 */
	public void sendBinaryMessage(ByteBuffer message) {
		if(message == null)
			throw new NullPointerException("Null message");
		
		ByteBuffer buffer = ByteBuffer.allocate(1 + message.remaining());
		buffer.put(MessageConstants.TYPE_BINARY);
		buffer.put(message);
		buffer.flip();
		sendMessage(buffer);
	}
	
	/**
	 * Sends the given object message to the server
	 * 
	 * @param message message to send
	 */
	public void sendObjectMessage(Message message) {
		if(message == null)
			throw new NullPointerException("Null message");
		
		byte[] serializedObject = ServerConfiguration.getObjectSerializer().serialize(message);
		
		ByteBuffer msg = ByteBuffer.allocate(1 + serializedObject.length);
		msg.put(MessageConstants.TYPE_OBJECT);
		msg.put(serializedObject);
		msg.flip();
		sendMessage(msg);
	}
	
	/**
	 * Sends a message to the server
	 */
	private void sendMessage(ByteBuffer message) {
		checkLoggedIn();

		MessageBuffer buf =
			new MessageBuffer(1 + message.remaining());
		buf.putByte(SimpleSgsProtocol.SESSION_MESSAGE).putBytes(message.array());
		try {
			connection.sendBytes(buf.getBuffer());
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}

	/**
	 * Retrieves the next binary session message or <code>null</code>
	 * if no messages are available.
	 */
	public ByteBuffer nextBinarySessionMessage() {
		synchronized (binarySessionMessageLock) {
			if (binarySessionMessages.isEmpty()) {
				try {
					binarySessionMessageLock.wait(WAIT_TIME * 2);
				} catch (InterruptedException e) {
				}
			}
			return (binarySessionMessages.isEmpty() ? null : binarySessionMessages.remove(0));
		}
	}

	/**
	 * Retrieves the next object session message or <code>null</code>
	 * if no messages are available.
	 */
	public Message nextObjectSessionMessage() {
		synchronized (objectSessionMessageLock) {
			if (objectSessionMessages.isEmpty()) {
				try {
					objectSessionMessageLock.wait(WAIT_TIME * 2);
				} catch (InterruptedException e) {
				}
			}
			return (objectSessionMessages.isEmpty() ? null : objectSessionMessages.remove(0));
		}
	}

	/**
	 * Sends a channel message to the server
	 */
	public void sendChannelMessage(String channelName, ByteBuffer message) {
		checkLoggedIn();
		sendChannelMessage(channelNameToId.get(channelName), message);
	}

	/**
	 * Internal helper method
	 */
	private void sendChannelMessage(BigInteger channelRefId, ByteBuffer message) {
		byte[] channelId = channelRefId.toByteArray();
		MessageBuffer buf =
			new MessageBuffer(3 + channelId.length + 4 + message.remaining());
		buf.putByte(SimpleSgsProtocol.CHANNEL_MESSAGE).
		putShort(channelId.length).
		putBytes(channelId).
		putInt(message.remaining()).
		putBytes(message.array());
		try {
			connection.sendBytes(buf.getBuffer());
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
	}

	/**
	 * Retrieves the next binary channel message or <code>null</code>
	 * if no messages are available.
	 */
	public BinaryChannelMessage nextBinaryChannelMessage() {
		synchronized (binaryChannelMessageLock) {
			if (binaryChannelMessages.isEmpty()) {
				try {
					binaryChannelMessageLock.wait(WAIT_TIME * 2);
				} catch (InterruptedException e) {
				}
			}
			return (binaryChannelMessages.isEmpty() ? null : binaryChannelMessages.remove(0));
		}
	}

	/**
	 * Retrieves the next object channel message or <code>null</code>
	 * if no messages are available.
	 */
	public ObjectChannelMessage nextObjectChannelMessage() {
		synchronized (objectChannelMessageLock) {
			if (objectChannelMessages.isEmpty()) {
				try {
					objectChannelMessageLock.wait(WAIT_TIME * 2);
				} catch (InterruptedException e) {
				}
			}
			return (objectChannelMessages.isEmpty() ? null : objectChannelMessages.remove(0));
		}
	}

	/**
	 * Waits for the client to join the given channel
	 */
	public void waitForJoin(String channelToJoin) {
		synchronized (lock) {
			try {
				if (channelNameToId.get(channelToJoin) == null)
					lock.wait(WAIT_TIME);

				if (channelNameToId.get(channelToJoin) == null)
					throw new RuntimeException("Channel join failed: " + channelToJoin);
			} catch (InterruptedException ex) {
				throw new RuntimeException("Channel join timed out: " + channelToJoin, ex);
			}
		}
	}

	/**
	 * Waits for the client to leave the given channel
	 */
	public void waitForLeave(String channelToLeave) {
		synchronized (lock) {
			try {
				if (channelNameToId.get(channelToLeave) != null)
					lock.wait(WAIT_TIME);

				if (channelNameToId.get(channelToLeave) != null)
					throw new RuntimeException("Channel leave failed: " + channelToLeave);
			} catch (InterruptedException ex) {
				throw new RuntimeException("Channel join timed out: " + channelToLeave, ex);
			}
		}
	}

	/**
	 * Returns client session
	 */
	public ClientSession getSession() {
		GetSessionTask task = new GetSessionTask(name);

		try {
			trxScheduler.runTask(task, owner);
			return task.getSession();
		} catch(Exception ex) {
			return null;
		}
	}

	/**
	 * {@inheritDoc}
	 */
	public void resetMessageQueues() {
		synchronized(binarySessionMessageLock) {
			binarySessionMessages.clear();
		}

		synchronized(objectSessionMessageLock) {
			objectSessionMessages.clear();
		}

		synchronized (binaryChannelMessageLock) {
			binaryChannelMessages.clear();
		}

		synchronized (objectChannelMessageLock) {
			objectChannelMessages.clear();
		}
	}

	private class GetSessionTask implements KernelRunnable {
		private final String name;
		private ClientSession session;

		GetSessionTask(String name) {
			this.name = name;
		}

		public void run() {
			session = (ClientSession)dataService.getBinding(PlayerService.PLAYER_PREFIX + name);
		}

		public ClientSession getSession() {
			return session;
		}

		public String getBaseTaskType() {
			return getClass().getName();
		}
	}

	private class Listener implements ConnectionListener {

		List<byte[]> messageList = new ArrayList<byte[]>();

		/** {@inheritDoc} */
		public void bytesReceived(Connection conn, byte[] buffer) {
			if (connection != conn) {
				throw new RuntimeException(
						"[" + name + "] bytesReceived: " +
						"wrong handle, got:" +
						conn + ", expected:" + connection);
			}

			MessageBuffer buf = new MessageBuffer(buffer);

			processAppProtocolMessage(buf);
		}

		private void processAppProtocolMessage(MessageBuffer buf) {

			byte opcode = buf.getByte();

			switch (opcode) {

			case SimpleSgsProtocol.LOGIN_SUCCESS:
				synchronized (lock) {
					loginAck = true;
					loginSuccess = true;
					lock.notifyAll();
				}
				break;

			case SimpleSgsProtocol.LOGIN_FAILURE:
				synchronized (lock) {
					loginAck = true;
					loginSuccess = false;
					lock.notifyAll();
				}
				break;

			case SimpleSgsProtocol.LOGOUT_SUCCESS:
				synchronized (lock) {
					logoutAck = true;
				}
				break;

			case SimpleSgsProtocol.LOGIN_REDIRECT:
				redirectHost = buf.getString();
				redirectPort = buf.getInt();
				synchronized (lock) {
					loginAck = true;
					loginRedirect = true;
					lock.notifyAll();
				} break;

			case SimpleSgsProtocol.SESSION_MESSAGE: {
				synchronized (lock) {
					byte[] msgBytes = buf.getBytes(buf.limit() - buf.position());
					ByteBuffer msg = ByteBuffer.wrap(msgBytes);

					int messageType = msg.get();
					if(messageType == MessageConstants.TYPE_BINARY) {
						synchronized(binarySessionMessageLock) {
							binarySessionMessages.add(msg.asReadOnlyBuffer());
							binarySessionMessageLock.notifyAll();
						}
					} else {
						synchronized(objectSessionMessageLock) {
							byte[] buffer = new byte[msg.remaining()];
							msg.get(buffer);

							Message deserializedMessage = 
								(Message)ServerConfiguration.getObjectSerializer().deserialize(buffer);
							objectSessionMessages.add(deserializedMessage);
							objectSessionMessageLock.notifyAll();
						}
					}

					lock.notifyAll();
				}
				break;
			}

			case SimpleSgsProtocol.CHANNEL_JOIN: {
				String channelName = buf.getString();
				BigInteger channelId = new BigInteger(1,
						buf.getBytes(buf.limit() - buf.position()));
				synchronized (lock) {
					channelIdToName.put(channelId, channelName);
					channelNameToId.put(channelName, channelId);
					lock.notifyAll();
				}
				break;
			}

			case SimpleSgsProtocol.CHANNEL_LEAVE: {
				BigInteger channelId = new BigInteger(1,
						buf.getBytes(buf.limit() - buf.position()));
				synchronized (lock) {
					channelIdToName.remove(channelId);
					lock.notifyAll();
				}
				break;

			}
			case SimpleSgsProtocol.CHANNEL_MESSAGE: {
				BigInteger channelId = new BigInteger(1,
						buf.getBytes(buf.getShort()));
				synchronized (lock) {
					String channelName = channelIdToName.get(channelId);

					byte[] msgBytes = buf.getBytes(buf.limit() - buf.position());
					ByteBuffer msg = ByteBuffer.wrap(msgBytes);

					int messageType = msg.get();
					if(messageType == MessageConstants.TYPE_BINARY) {
						synchronized(binaryChannelMessageLock) {
							binaryChannelMessages.add(new BinaryChannelMessage(channelName, msg.asReadOnlyBuffer()));
							binaryChannelMessageLock.notifyAll();
						}
					} else {
						synchronized(objectChannelMessageLock) {
							byte[] buffer = new byte[msg.remaining()];
							msg.get(buffer);

							Message deserializedMessage = 
								(Message)ServerConfiguration.getObjectSerializer().deserialize(buffer);
							objectChannelMessages.add(new ObjectChannelMessage(channelName, deserializedMessage));
							objectChannelMessageLock.notifyAll();
						}
					}

					lock.notifyAll();
				}
				break;
			}

			default:
				System.err.println( 
						"[" + name + "] " +
						"processAppProtocolMessage: unknown op code: " +
						opcode);
			break;
			}
		}

		/** {@inheritDoc} */
		public void connected(Connection conn) {
			if (connection != null)
				throw new RuntimeException("Client already connected");

			connection = conn;
			synchronized (lock) {
				connected = true;
				lock.notifyAll();
			}
		}

		/** {@inheritDoc} */
		public void disconnected(Connection conn) {
			synchronized (lock) {
				if (awaitGraceful) {
					// Hack since client might not get last msg
					logoutAck = true;
				}

				try {
					trxScheduler.runTask(new KernelRunnable() {
						public void run() {
							// remove client session from the store
							dataService.removeBinding(PlayerService.PLAYER_PREFIX + name);
						}

						public String getBaseTaskType() {
							return getClass().getName();
						}
					}, owner);
				} catch(Exception ex) {
					// TODO: add handling
				}

				reset();
				lock.notifyAll();
			}
		}

		/** {@inheritDoc} */
		public void exceptionThrown(Connection conn, Throwable exception) {
			System.err.println("Exception thrown in listener:" + exception);
			exception.printStackTrace();
		}

	}

    /**
     * Raw channel message wrapper class
     */
    public static class BinaryChannelMessage {
        private String channelName;
        private ByteBuffer message;
        
        public BinaryChannelMessage(String channelName, ByteBuffer message) {
            this.channelName = channelName;
            this.message = message;
        }
        
        public String getChannelName() {
            return channelName;
        }
        
        public ByteBuffer getMessage() {
            return message;
        }
    }

    /**
     * Raw channel message wrapper class
     */
    public static class ObjectChannelMessage {
        private String channelName;
        private Message message;
        
        public ObjectChannelMessage(String channelName, Message message) {
            this.channelName = channelName;
            this.message = message;
        }
        
        public String getChannelName() {
            return channelName;
        }
        
        public Message getMessage() {
            return message;
        }
    }

}
